#include "threadpool.h"

#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <thread>

#include "task.h"

const int TASK_MAX_THRESHHOLD = INT32_MAX;
const int THREAD_MAX_THRESHHOLD = 1024;
const int THREAD_MAX_IDLE_TIME = 60;  // 单位：秒

int Thread::generateId_ = 0;
Thread::Thread(ThreadFunc func) : func_(func), threadId_(generateId_++) {
}

Thread::~Thread() {
    // do nothing
}

void Thread::start() {
    std::thread t(func_, threadId_);
    t.detach();
}

ThreadPool::ThreadPool()
    : initThreadSize_(0),
      threadSizeThreshhold_(THREAD_MAX_THRESHHOLD),
      idleThreadSize_(0),
      taskSizeThreshhold_(TASK_MAX_THRESHHOLD),
      mode_(PoolMode::MODE_FIXED),
      isPoolRunning_(false) {
}

ThreadPool::~ThreadPool() {
    isPoolRunning_ = false;

    std::unique_lock<std::mutex> lock(mtx_);
    notEmpty_.notify_all();
    exitCond_.wait(lock, [&]() -> bool { return threads_.size() == 0; });
}

void ThreadPool::setMode(PoolMode mode) {
    if (checkRunningState()) return;
    mode_ = mode;
}
void ThreadPool::setTaskSizeThreshhold(int threshhold) {
    if (checkRunningState()) return;
    taskSizeThreshhold_ = threshhold;
}
void ThreadPool::setThreadSizeThreshhold(int threshhold) {
    if (checkRunningState()) return;
    threadSizeThreshhold_ = threshhold;
}

Result ThreadPool::submit(std::shared_ptr<Task> task) {
    std::unique_lock<std::mutex> lock(mtx_);
    if (!notFull_.wait_for(lock, std::chrono::seconds(1), [&]() -> bool {
            return tasks_.size() < taskSizeThreshhold_;
        })) {
        // 等待超时 条件依然不满足
        std::cerr << "[Error] Task queue if full, failed to submit."
                  << std::endl;
        return {task, false};  // invalid result
    }

    tasks_.emplace(task);
    notEmpty_.notify_all();

    if (mode_ == PoolMode::MODE_CACHED && tasks_.size() > idleThreadSize_ &&
        threads_.size() < threadSizeThreshhold_) {
        std::cout << "[Info] creating new thread..." << std::endl;

        auto ptr = std::make_unique<Thread>(
            std::bind(&ThreadPool::threadfunc, this, std::placeholders::_1));
        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));

        threads_[threadId]->start();
        idleThreadSize_++;
    }

    return {task};
}

void ThreadPool::start(int initThreadSize) {
    isPoolRunning_ = true;

    initThreadSize_ = initThreadSize;
    for (int i = 0; i < initThreadSize_; ++i) {
        auto ptr = std::make_unique<Thread>(
            std::bind(&ThreadPool::threadfunc, this, std::placeholders::_1));
        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));
    }

    for (int i = 0; i < initThreadSize_; ++i) {
        threads_[i]->start();
        idleThreadSize_++;
    }
}

void ThreadPool::threadfunc(int threadId) {
    auto lastTime = std::chrono::high_resolution_clock().now();
    for (;;) {
        std::shared_ptr<Task> task;
        {
            std::unique_lock<std::mutex> lock(mtx_);
            std::cout << "[Info] tid: " << std::this_thread::get_id()
                      << " try to get task." << std::endl;
            while (tasks_.size() == 0) {
                if (!isPoolRunning_) {
                    threads_.erase(threadId);
                    std::cout << "[Info] tid: " << std::this_thread::get_id()
                              << " exit!" << std::endl;

                    exitCond_.notify_all();
                    return;
                }

                if (mode_ == PoolMode::MODE_CACHED) {
                    if (std::cv_status::timeout ==
                        notEmpty_.wait_for(lock, std::chrono::seconds(1))) {
                        // 等待超时, 自动唤醒
                        auto now = std::chrono::high_resolution_clock().now();
                        auto duration =
                            std::chrono::duration_cast<std::chrono::seconds>(
                                now - lastTime);

                        if (threads_.size() > initThreadSize_ &&
                            duration.count() >= THREAD_MAX_IDLE_TIME) {
                            threads_.erase(threadId);
                            idleThreadSize_--;

                            std::cout
                                << "[Info] tid: " << std::this_thread::get_id()
                                << " exit!" << std::endl;
                            return;
                        }
                    }
                } else {
                    notEmpty_.wait(lock);  // 不忙等
                }
            }

            idleThreadSize_--;
            std::cout << "[Info] tid: " << std::this_thread::get_id()
                      << " retrieve task successfully." << std::endl;

            task = tasks_.front();
            tasks_.pop();

            if (tasks_.size() > 0) {
                notEmpty_.notify_all();
            }
            notFull_.notify_all();
        }

        if (task) task->exec();
        idleThreadSize_++;
        lastTime = std::chrono::high_resolution_clock().now();
    }
}
